package com.hu.flink12.sensor

import com.hu.entity.SensorReading
import org.apache.flink.api.common.RuntimeExecutionMode
import org.apache.flink.streaming.api.scala._

/**
 * @Author: hujianjun
 * @Create Date: 2021/1/26 16:01
 * @Describe: 拆分筛选流
 */

object UnionAndConnectSensorStream {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val inputStream1 = env.readTextFile(getClass.getResource("/data/sensor.csv").getPath, "utf-8")
    val inputStream2 = env.readTextFile(getClass.getResource("/data/sensor.txt").getPath, "utf-8")

    val dataStream1 = inputStream1.map(data => {
      val str = data.split(",")
      SensorReading(str(0), str(1).trim.toLong, str(2).trim.toDouble)
    })

    val dataStream2 = inputStream2.map(data => {
      val str = data.split(",")
      SensorReading(str(0), str(1).trim.toLong, str(2).trim.toDouble)
    })

    println(dataStream1.connect(dataStream2))
    dataStream1.union(dataStream2).union(dataStream1).print("::union")

    env.setRuntimeMode(RuntimeExecutionMode.BATCH)
    env.execute("读取传感器csv数据")
  }
}
